/*
 Copyright (C) 2011  Matteo Magnani, Gabriele Nunziante
 
 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation, either version 3 of the License, or
 (at your option) any later version.
 
 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.
 
 You should have received a copy of the GNU General Public License
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

/************************************************************************************
 * CRS : Server to retrieve trending topics and distribute them to Tweet Retrievers *
 ************************************************************************************
 * 
 ************************************************************************************/


import java.io.*;
import java.net.*;
import java.util.*;
import java.text.DateFormat;
import java.text.SimpleDateFormat;

import javax.xml.parsers.*;

import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.json.simple.JSONValue;
import org.w3c.dom.*;


public class CRTrendServer extends Thread {
	
	/* DEBUG variables */
	boolean s = false; /* true to print system messages */
	boolean t = false; /* true to print tags */
	
	private ServerSocket Server;
	static int port, n_tag;
	static TrendArray trendArray;
	private String keywordFile = "keywords";
	
	private static String url_trend = "http://api.twitter.com/1/trends/23424853.json";
	
	/**
	 * @param args[0] server port
	 * @param args[1] number of tags to retrieve each time (from 1 to 10)
	 */
	public static void main(String args[]) throws Exception {
		port = Integer.parseInt(args[0]);
		n_tag = Integer.parseInt(args[1]);
		trendArray = new TrendArray(10000);
		
		if( n_tag <= 0 || n_tag > 10 )
			System.out.println("Number of tags not valid: specify a number between 1 and 10");
		else {
			new CRTrendServer();
		}
	}

	public CRTrendServer() throws Exception {
		/* Reads keywords */
		LineNumberReader reader = new LineNumberReader(new FileReader(keywordFile));
		String line;
		while ((line=reader.readLine())!=null) {
			String mid = findMaxID(line);
			if (!mid.equals("-1"))
				trendArray.tryAddObject(line, mid);
		}
		reader.close();
		
		
		/* Gets current trends */
		URL u = new URL(url_trend);
		InputStream is = u.openStream();
		BufferedReader dis = new BufferedReader( new InputStreamReader(is) );
	
		String str, message = "";
		while ((str = dis.readLine()) != null) {
			message += str;
		}
		
		String[] queries = getQueries(message);
		
		if (s) System.out.println("[CRTrendServer] Current Trends: \n"+queries+"\n");
		
		/* Inserts trending tags into the array */
		trendArray.resetActive();
		for (String query: queries) {
			trendArray.tryAddObject(query, findMaxID(query));
		}
		
		if (trendArray.size()==0) {
			System.err.println("[CRTrendServer] Cannot fill trend array");
			System.exit(-1);
		}
				
				
		/* Waits for incoming connections */
		Server = new ServerSocket(port);
		System.out.println("[CRS] *** Trend Server started on port " + port + " ***");
		this.start();
		
		/* The main thread waits 7 minutes, then it gets new trends and updates the TrendArray */
		while (true) {
			if (t) System.out.println(now() + "\n" + trendArray.getTrendArray());
			Thread.sleep(60000*5);
			
			try {
			if (s) System.out.println("[CRTrendServer] New request for current trends");
			
			/* Inserts keywords into the array */
			trendArray.resetActive();
			reader = new LineNumberReader(new FileReader(keywordFile));
			while ((line=reader.readLine())!=null) {
				String mid = findMaxID(line);
				if (!mid.equals("-1"))
					trendArray.tryAddObject(line, mid);
			}
			reader.close();
			
			u = new URL(url_trend);
			is = u.openStream();
			dis = new BufferedReader( new InputStreamReader(is) );
		
			message = "";
			while ((str = dis.readLine()) != null) {
				message += str;
			}
			
			queries = getQueries(message);
			
			/* Inserts trending tags into the array */
			trendArray.resetActive();
			for (String query: queries) {
				trendArray.tryAddObject(query, findMaxID(query));
			}
				
			}
			catch (Exception e) {
				System.err.println("[CRTrendServer] " + now() + " Cannot get new trending topics");
				e.printStackTrace();
			}
		}
	}

	private String[] getQueries(String r) {
		JSONArray array = (JSONArray)JSONValue.parse(r);
		JSONArray trends = (JSONArray)((JSONObject)array.get(0)).get("trends");
		ArrayList<String> res = new ArrayList<String>();
		for (int i=0; i<trends.size(); i++) {
			String v = (String)((JSONObject)trends.get(i)).get("query");
			if (v!=null)
				res.add(v);
		}
		return res.toArray(new String[0]);
	}
	
	/*  */
	public void run() {
		int cont = 0;
		String tag, maxid, minid;
		
		while(true) {
			try {
				Socket client = Server.accept();
				
				if (s) System.out.println("[CRTrendServer] Processing connection from " + client.getInetAddress());
				
				boolean continueProcessing = false;
				do {
					/* gets the first Tag to be searched */
					cont = trendArray.getActive(cont);
								
					tag = trendArray.getTag(cont);
					maxid = trendArray.getCount(cont);
					minid = trendArray.getMin(cont);
					
					if ( Long.parseLong(maxid) <= Long.parseLong(minid) ) {
						System.out.println("[CRTrendServer] MaxID < MinID! ["+maxid+" < "+minid+"]");
						
						trendArray.freezeTag(tag);
						//maxid = trendArray.getCount(cont);
						//minid = trendArray.getMin(cont);
						continueProcessing = true;
					}
					else
						continueProcessing = false;
				
				} while (continueProcessing);
				
				/* Sends to the client the search to be performed */
				Connect c = new Connect(this, client, tag, maxid, cont);
				
				/* selects the next element of the array */
				cont = (cont == (trendArray.size()-1) ? 0 : ++cont);
			}
			catch(Exception e) { 
				System.err.println("[CRTrendServer] " + now() + " Error processing connection");
				e.printStackTrace(); 
			}
		}
	}

	
	/**
	 * @return The highest tweet ID for a Tag
	 */
	private synchronized String findMaxID(String tag_s) throws Exception {
		String tag = tag_s;
		String id = "";
		
		tag = tag.replaceAll("#", "%23");
		tag = tag.replaceAll(" ", "%20");
		
		String search = "http://search.twitter.com/search.atom?q=" + tag + "&rpp=1";

		Document doc = getDocumentFromURL( search );

		NodeList friends = doc.getElementsByTagName("entry");
		
		if (friends.getLength()==0) return "-1";
		
		Node userNode = friends.item( friends.getLength()-1 );
		NodeList children = userNode.getChildNodes();

		for (int j=0; j<children.getLength(); j++)
			if (children.item(j).getNodeName().equals("id"))
				id = children.item(j).getFirstChild().getNodeValue();

		id = id.replaceAll("tag:search.twitter.com,2005:", "");

		if (s) System.out.println("[CRTrendServer] MaxID for Tag "+tag+" = "+id);
		return id;
	}

	
	private static Document getDocumentFromURL(String url) throws Exception {
		URLConnection conn = new URL(url).openConnection();
		conn.setConnectTimeout(10000);
		conn.setReadTimeout(10000);
		
		DocumentBuilder parser = DocumentBuilderFactory.newInstance().newDocumentBuilder();
		Document doc = parser.parse( conn.getInputStream() );
		return doc;
	}
	
	private String now() {
		String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
		Calendar cal = Calendar.getInstance();
		SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMAT);
		return sdf.format(cal.getTime());
	}
}




/* Class used to manage single connections */
class Connect extends Thread {	
	
	/* DEBUG variables */
	boolean s = false; /* true to print system messages */
	
	private final CRTrendServer parent;
	private Socket client = null;
	private String tag, max_id;
	private int cont;
	BufferedReader in = null;
	PrintStream out = null;
		
	public Connect(CRTrendServer parentServer, Socket clientSocket, String hashtag, String maxid, int cont_id) {
		parent = parentServer;
		client = clientSocket;
		tag = hashtag;
		max_id = maxid;
		cont = cont_id;

		try {
			in = new BufferedReader(new InputStreamReader(client.getInputStream()));
			out = new PrintStream(client.getOutputStream(), true);
		}
		catch(Exception e1)
		{
			try { client.close(); }
			catch(Exception e) { System.out.println(e.getMessage());}
			return;
		}
		
		this.start();
	}
	
	
	public void run() {
		try {
			/* Normalizes # and spaces before using them in url */
			tag = tag.replaceAll("#", "%23");
			tag = tag.replaceAll(" ", "%20");
			
			String answer = tag + "\t" + max_id;
			if (s) System.out.println("[TrendServer] new request for " + client.getInetAddress() + " -> " + answer);

			out.println(answer);
			out.flush();
			
			
			/* ID of last dowloaded tweet */
			String message = "", message2 = "";
			while ( (message2 = in.readLine()) != null ) {
				message += message2;
			}
			
			if (s) System.out.println("[TrendServer] last tweet ID from " + client.getInetAddress() + " - Tag \"" + tag + "\": " + message);
			if ( !message.equals("end") ) {
				/* next search on this Tag will start from this ID */
				try {
					Long.parseLong(message);
					parent.trendArray.setCount(cont, message);
				} catch (Exception e) {
					System.err.println("[TrendServer] " + now() + " " +  client.getInetAddress() + ": last ID for Tag \""+tag+"\" not valid [" + message + "] " + e.getMessage());
					parent.trendArray.setMin(cont);
				}
			}
			else {
				/* If no ID is retrieved, we set the lower bound */
				if (s) System.out.println("[TrendServer] last ID for Tag \""+tag+"\" not received: setting lower bound");
				parent.trendArray.setMin(cont);
			}
			
			out.close();
			in.close();
			client.close();
		}
		catch(Exception e) { 
			e.printStackTrace(); 
		}
	}
	
	private String now() {
		String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
		Calendar cal = Calendar.getInstance();
		SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMAT);
		return sdf.format(cal.getTime());
	}
}

/* Class used to keep metadata about trending tags */
class TrendArray {
	
	/* DEBUG variables */
	boolean s = false; /* true to print system messages */
	boolean t = false; /* true to print new tags */
	
	/* Status of the tag */
	private static String USED = "2";
	private static String ACTIVE = "1";
	private static String INACTIVE = "0";
	private static String FREEZE = "-1";
	
	/* Maximum number of tags stored in this array */
	private int maxSize = 100;
	private int numDeletions = 5;
	
	Vector<String> tag_array;
	Vector<String> max_id;
	Vector<String> count_id;
	Vector<String> min_id;
	Vector<String> active;
	
	public TrendArray() {
		tag_array = new Vector<String>();
		max_id = new Vector<String>();
		count_id = new Vector<String>();
		min_id = new Vector<String>();
		active = new Vector<String>();
	}
	
	public TrendArray(int maxSize) {
		this();
		this.maxSize = maxSize;
	}
	
	/** 
	 * Adds a tag to the end of the array.
	 * @param tag Tag to be added
	 * @param max max tweet identifier already retrieved 
	 */
	public int addObject(String tag, String max) {
		if (tag_array.size()>=maxSize) {
			for (int i=0; i<numDeletions; i++) {
				tag_array.remove(0);
				max_id.remove(0);
				count_id.remove(0);
				min_id.remove(0);
				active.remove(0);
			}
		}
			
		tag_array.add( tag );
		max_id.add( max );
		count_id.add( max );
		min_id.add( "0" );
		active.add( ACTIVE );
		//if (t) System.out.println("[TrendArray] [" + now() + "] new tag - " + tag_array.size() + "["+tag+"]");
		
		return (tag_array.size()-1);
	}
	
	/*
	 * Performed when all tweets related to this Tag have been downloaded
	 */
	public void freezeTag(String tag) {
		for (int i=0; i<tag_array.size(); i++) {
			if ( tag_array.get(i).equals(tag) )
				active.set(i, FREEZE);
		}
	}
		
	/*
	 * Sets all Tags as INACTIVE
	 */
	public void resetActive() {
		for (int i=0; i<tag_array.size(); i++) {
			if ( !active.get(i).equals( FREEZE ) )
				active.set(i, INACTIVE);
		}
	}
	
	
	/**
	 * Adds or activate a tag to the array. If the tag is already present it is set as ACTIVE, otherwise a new tag is inserted.
	 * @param tag
	 * @param max
	 **/
	public synchronized int tryAddObject(String tag, String max) {
		if ( tag_array.contains(tag) ) {
			for (int i=0; i<tag_array.size(); i++) {
				if (tag_array.get(i).equals(tag)) {
					if ( active.get(i).equals( FREEZE ) ) {
						min_id.set(i, max_id.get(i));
						max_id.set(i, max);
						count_id.set(i, max );
					}
					active.set(i, ACTIVE);
				}
			}
			if (s) System.out.println("[TrendArray] Tag \""+tag+"\" set as ACTIVE");
		}
		else {
			tag_array.add( tag );
			max_id.add( max );
			count_id.add( max );
			min_id.add( "0" );
			active.add( ACTIVE );
			if (s) System.out.println("[TrendArray] Tag \""+tag+"\" ADDED");
			//if (t) System.out.println("[TrendArray] [" + now() + "] new tag - " + tag_array.size() + "["+tag+"]");
		}
		
		
		/* notifies the availability of a new tag */
		try {
			notify();
		} catch (Exception e) {e.printStackTrace();}
		
		return (tag_array.size()-1);
	}
	
	
	/*
	 * @return The index of the first active tag in the array
	 * @param cont Starting index from which the Tag is searched
	 */
	public synchronized int getActive(int cont) {
		int i = cont;
		int k = 0;
		
		while ( active.get(i) != ACTIVE && k < tag_array.size() ) {
			k++;
			i++;
			if( i >= active.size())
				i = 0;
		}
		
		/* Tag not found: pauses */
		if( k >= tag_array.size()) {
			try {
				wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			
			/* notify() received: new Tag available */
			i = cont;
			k = 0;
			while( active.get(i) != ACTIVE && k < tag_array.size() ) {
				k++;
				i++;
				if( i >= active.size())
					i = 0;
			}
		}
		
		/* Updates tag status to USED */
		active.set(i, USED);
		
		return i;
	}
	
	public String getTag(int cont) {
		return tag_array.get(cont);
	}
	
	public String getCount(int cont) {
		return count_id.get(cont);
	}
	

	public String getMin(int cont) {
		return min_id.get(cont);
	}
	
	public synchronized void setCount(int cont, String value) {
		count_id.set(cont, value);
		
		if( active.get(cont).equals( USED ))
			active.set(cont, ACTIVE);

		try {
			notify();
		} catch(Exception e) {}
	}
	
	public void setMin(int cont) {
		min_id.set(cont, count_id.get(cont));
		active.set(cont, FREEZE);
	}
	
	public int size() {
		return tag_array.size();
	}
	
	public String getTrendArray() {
		String message = "TAG \t\tMAX ID \tCOUNT ID \tMIN ID \tACTIVE\n";
		for(int i=0; i<tag_array.size(); i++)
		{
			message += tag_array.get(i) + " \t";
			message += max_id.get(i) + " \t";
			message += count_id.get(i) + " \t";
			message += min_id.get(i) + " \t";
			message += active.get(i) + " \n";
		}
		
		return message;
	}
	
	private String now() {
		String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
		Calendar cal = Calendar.getInstance();
		SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMAT);
		return sdf.format(cal.getTime());
	}
}